package com.doit.day07

import org.apache.spark.sql.{Encoder, Encoders}
import org.apache.spark.sql.expressions.Aggregator

/**
 * @Author: Hang.Nian.YY
 * @WX: 17710299606
 * @Tips: 学大数据 ,到多易教育
 * @DOC: https://blog.csdn.net/qq_37933018?spm=1000.2115.3001.5343
 * @Description:
 * max 最大年龄
 * 泛型1  输入    年龄  Int
 * 泛型2  计算的中间缓存
 * 泛型3  输出结果
 */


class MyUdaf extends Aggregator[Int, Int, Int] {
  var buff: Int = 0

  // 在每个分区任务重初始化一次 缓存
  override def zero: Int = {
    buff = 0
    buff
  }

  // 分区内部的计算逻辑
  override def reduce(b: Int, a: Int): Int = {
    buff = List(buff, a, b).max
    buff
  }

  // 分区间的聚合
  override def merge(b1: Int, b2: Int): Int = {
    buff = List(buff, b1, b2).max
    buff
  }
// 返回最终结果
  override def finish(reduction: Int): Int = reduction

  // 序列化使用的数据类型
  override def bufferEncoder: Encoder[Int] = Encoders.scalaInt
  override def outputEncoder: Encoder[Int] = Encoders.scalaInt
}
